package com.my.study.listener;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.my.study.config.CanalConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.List;

@Component
@Slf4j
public class CanalListener implements CommandLineRunner{

    @Autowired
    private CanalConfig canalConfig;

    /**
     * 开始日志监控
     */
    @Override
    public void run(String... args) {
        // 创建链接
        CanalConnector connector = CanalConnectors
                .newSingleConnector(new InetSocketAddress(canalConfig.getCanalServerIp(),
                        canalConfig.getCanalServerPort()), canalConfig.getDestination(),
                        canalConfig.getUserName(), canalConfig.getPassword());
        try {
            connector.connect();
            // canal配置文件中配置
            connector.subscribe(canalConfig.getTableName());
            connector.rollback();
            log.info(">>>>>>>>>>>>>>>>开始监听>>>>>>>>>>>>>>");
            while (true) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(canalConfig.getBatchSize());
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    try {
                        log.info(">>>>>>>>>>>>>>>>无变化>>>>>>>>>>>>>>");
                        Thread.sleep(10000);
                    } catch (InterruptedException e) {
                    }
                } else {
                    printEntry(message.getEntries());
                }
                // 提交确认
                connector.ack(batchId);
            }
        } finally {
            connector.disconnect();
        }
    }

    private static void printEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }

            CanalEntry.RowChange rowChage = null;
            try {
                rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }

            CanalEntry.EventType eventType = rowChage.getEventType();

            for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == CanalEntry.EventType.DELETE) {
                    // 删除，获取删除前的数据
                    log.info("delete data");
                    //redisDelete(rowData.getBeforeColumnsList(),entry.getHeader().getTableName());
                } else if (eventType == CanalEntry.EventType.INSERT) {
                    // 新增，获取新增后的数据
                    log.info("insert data");
                    //redisSet(rowData.getAfterColumnsList(),entry.getHeader().getTableName());
                } else {
                    // 修改，获取修改后的数据
                    log.info("update data");
                    // printColumn(rowData.getBeforeColumnsList());
                    //redisSet(rowData.getAfterColumnsList(),entry.getHeader().getTableName());
                }
            }
        }
    }

    /*private static void redisSet(List<CanalEntry.Column> columns, String tableName) {
        JSONObject json = new JSONObject();
        // 字段和值的列表，放入json，后续作为redis的值
        for (CanalEntry.Column column : columns) {
            json.put(column.getName(), column.getValue());
        }
        if (columns.size() > 0) {
            System.out.println("set key: " + columns.get(0).getValue() + ", value: " + json.toJSONString());
            // 获取第一列主键的数据
            RedisUtil.setString(tableName+":" + columns.get(0).getValue(), json.toJSONString());
        }
    }

    private static void redisDelete(List<CanalEntry.Column> columns, String tableName) {
        JSONObject json = new JSONObject();
        for (CanalEntry.Column column : columns) {
            json.put(column.getName(), column.getValue());
        }
        if (columns.size() > 0) {
            System.out.println("delete key: " + columns.get(0).getValue());
            RedisUtil.del(tableName+":" + columns.get(0).getValue());
        }
    }*/
}
